package com.zyh.kafka.day02;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

public class MyProducerInterceptor implements ProducerInterceptor<String,String> {
    // 成功次数
    private long successNum = 0;
    // 失败次数
    private long errorNum = 0;
    /**
     * 在消息发送到kafka之前执行，可以对发送的所有消息进行预处理
     * @param producerRecord 消息对象
     * @return 消息对象
     */
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
        long timestamp = System.currentTimeMillis();
        String value = producerRecord.value();
        value = timestamp+"-"+ value;

        return new ProducerRecord<String,String>(producerRecord.topic(), producerRecord.partition(), producerRecord.key(), value, producerRecord.headers());
    }

    /**
     * 响应到达producer之前执行的方法，通常可以统计发送成功和失败的消息格式
     * @param recordMetadata 消息元数据
     * @param e 异常
     */
    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
        if(e!=null){
            errorNum++;
            System.out.println(errorNum);
        }else{
            successNum++;
            System.out.println(successNum);
        }
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}
